package rabbit.client.custom;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class MyKafkaConsumer {
	public static void main(String[] args) {
		//配置信息
		Properties props = new Properties();
		props.put ("bootstrap.servers","localhost:9092");
		props.put ("group.id","test");
		//设置数据 key 的序列化处理类
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		//设置数据 value 的序列化处理类
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
		consumer.subscribe(Arrays.asList("my-topic"));
		while(true) {
			ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
			for(ConsumerRecord<String,String> record:records) {
				System.out.println("key==>"+record.key()+"    value===>"+record.value());
			}
		}
		 
	}
}
